#include <sys/types.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <errno.h>
#include <assert.h>
#include <fcntl.h>
#include <string.h>
#include <pthread.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <event2/event_struct.h>
#include <event2/event.h>
#include <arpa/inet.h>

#include "utils.h"
#include "common.h"
#include "hashtable.h"
#include "myrbtree.h"

void tracker_handle_cmd(int sockfd,short event,void *arg);

struct storage_cluster
{
    rb_red_blk_tree *sgrb_tree;
    int num;
};
struct thread_pool
{
    int size;
    struct thread_worker *threads;
};

struct tracker_struct
{
    int clifd;
    int chkfd;
    struct event cli_ev;
    struct event chk_ev;
    struct event_base *main_base;
    struct thread_pool *pool;
    struct storage_cluster sg_cluster; 
    struct hash_table *htable;
    struct hash_table *do_htable;
};

struct buffer
{
    char *buf;
    int pos;
    int length;
};
struct connection
{
    list_node node;
    int fd;
    enum CH_STAT stat;
    struct event ev;
    struct buffer rdbuf;
    struct buffer wrbuf;
    struct thread_worker *worker;

};


struct thread_worker
{
    pthread_t pid;
    int notify_read;
    int notify_write;
    struct event notify_ev;
    struct event_base *base;
    struct tracker_struct *tracker;
    pthread_mutex_t mutex;
    list_node connections;
};

int client_network_init(struct tracker_struct *traker);
int chunker_network_init(struct tracker_struct *traker);
int thread_pool_init(struct tracker_struct *tracker,int size);
void tracker_accept(int sockfd, short event, void* arg);
void init_environment(void);

void delete_storage(struct storage_cluster *sc,int storageid)
{
    rb_red_blk_node *node = NULL;

    node = RBExactQuery(sc->sgrb_tree,&storageid);
    RBDelete(sc->sgrb_tree,node);
    return;
}

void insert_storage(struct storage_cluster *sc,struct storage *storage)
{
    int *key = (int *)malloc(sizeof(int));
    if (key == NULL) {
        return;
    }
    RBTreeInsert(sc->sgrb_tree,key,storage);
    sc->num++;
    return;
}

int register_chunker(struct tracker_struct *tracker,struct storage *storage)
{
    struct storage_cluster *sc = &(tracker->sg_cluster);
    insert_storage(sc,storage);
    return 0;         
}

int dfs_hashtable_init(struct hash_table *htable)
{
    if ((htable = hash_create(4096)) == NULL) {
        printf("create hashtable failed\n");
        return -1;
    }
    return 0;
}

int storage_cluster_init(struct storage_cluster *cluster)
{
    cluster->num = 0;
    cluster->sgrb_tree = create_storage_rbtree();
    return 0;
}

int main(int argc,char **argv)
{
    init_environment();
    struct tracker_struct tracker;
    tracker.main_base = event_base_new();
    thread_pool_init(&tracker,0);
    client_network_init(&tracker);
    chunker_network_init(&tracker);
    dfs_hashtable_init(tracker.htable);
    dfs_hashtable_init(tracker.do_htable);
    storage_cluster_init(&tracker.sg_cluster);


    event_base_dispatch(tracker.main_base);
    printf("tracker\n");
    return 0;
}

void init_environment()
{
#if 0
    int error;
    struct rlimit limit;
    limit.rlim_cur = 4096;
    limit.rlim_max = RLIM_INFINITY;

    if ((error = setrlimit(RLIMIT_NOFILE,&limit)) < 0) {
        printf("setrlimt:%s\n",strerror(errno));
    }
#endif
    struct rlimit rlim;
    int status = getrlimit(RLIMIT_NOFILE, &rlim);
    if (status == 0) {
        rlim.rlim_cur = rlim.rlim_max;
        setrlimit(RLIMIT_NOFILE, &rlim);
    }
    return;
}

struct thread_worker *get_thread_worker(struct tracker_struct *tracker,int sockfd)
{
    int i;
    struct thread_worker *worker;
    i = sockfd%(tracker->pool->size);

    worker = &tracker->pool->threads[i];
    return worker;
}


void tracker_buffer_init(struct buffer *buf,int size)
{
    buf->buf = malloc(1024);
    assert(buf->buf);
    buf->length = 1024;
    buf->pos = 0;
}
struct connection *connection_new(int fd)
{
    struct connection *conn;
    int flags = 0;
    conn = malloc(sizeof(struct connection));
    assert(conn);
    conn->fd = fd;
    flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
    tracker_buffer_init(&conn->rdbuf,0);
    tracker_buffer_init(&conn->wrbuf,0);
    return conn;
}

void tracker_disable_event(int sockfd,struct tracker_struct *tracker)
{
    if (sockfd == tracker->clifd) {
        event_set(&tracker->cli_ev, sockfd, 0, tracker_accept, tracker);
        event_base_set(tracker->main_base, &tracker->cli_ev);
        event_add(&tracker->cli_ev, NULL);
    }else {
        event_set(&tracker->chk_ev, sockfd, 0, tracker_accept, tracker);
        event_base_set(tracker->main_base, &tracker->chk_ev);
        event_add(&tracker->chk_ev, NULL);
    }
    return;
}

void tracker_accept(int sockfd, short event, void* arg)
{
    struct sockaddr_in client;
    int len;
    struct tracker_struct *tracker;
    int acceptfd;
    struct thread_worker *worker;
    struct connection *entry;
    int ret = 0;

    tracker = arg;
    acceptfd = accept(sockfd,(struct sockaddr*)&client,&len);
    printf("accept fd is %d\n",acceptfd);
    if (acceptfd < 0) {
        if (errno == EAGAIN) {
        }
        if (errno == EMFILE) {
            printf("%s\n",strerror(errno));
        }
        tracker_disable_event(sockfd,tracker);
        return;
    }
    worker = get_thread_worker(tracker,acceptfd);
    entry = connection_new(acceptfd);
    pthread_mutex_lock(&worker->mutex);
    list_insert_after(&worker->connections,&entry->node),
    pthread_mutex_unlock(&worker->mutex);

    ret = write(worker->notify_write,"c",1);
    return;
}

int client_network_init(struct tracker_struct *tracker)
{
    int sockfd = socket(AF_INET,SOCK_STREAM,0);
    struct sockaddr_in server;
    int acceptfd;
    struct sockaddr_in client;
    int len;

    server.sin_family = AF_INET;
    server.sin_port = htons(CLIPORT);
    server.sin_addr.s_addr = INADDR_ANY;
    if (sockfd < 0) {
    //log(LOG_ERR,"%s %s:create socket failed",__FUNCTION__,__LINE__);
        return -1;
    }

    if (bind(sockfd,(struct sockaddr*)&server,sizeof(server)) < 0) {
        close(sockfd);
        return -1;
    }

    if (listen(sockfd,1024) < 0) {
        close(sockfd);
        return -1;
    }

    tracker->clifd = sockfd;
    event_set(&tracker->cli_ev, sockfd, EV_READ|EV_PERSIST, tracker_accept, tracker);
    event_base_set(tracker->main_base, &tracker->cli_ev);
    event_add(&tracker->cli_ev, NULL);
    return 0;
}
int chunker_network_init(struct tracker_struct *tracker)
{
    int sockfd = socket(AF_INET,SOCK_STREAM,0);
    struct sockaddr_in server;
    int acceptfd;
    struct sockaddr_in client;
    int len;

    server.sin_family = AF_INET;
    server.sin_port = htons(CHKPORT);
    server.sin_addr.s_addr = INADDR_ANY;
    if (sockfd < 0) {
    //log(LOG_ERR,"%s %d:create socket failed",__FUNCTION__,__LINE__);
        return -1;
    }

    if (bind(sockfd,(struct sockaddr*)&server,sizeof(server)) < 0) {
        close(sockfd);
        return -1;
    }

    if (listen(sockfd,1024) < 0) {
        close(sockfd);
        return -1;
    }

    tracker->chkfd = sockfd;
    event_set(&tracker->chk_ev, sockfd, EV_READ|EV_PERSIST, tracker_accept, tracker);
    event_base_set(tracker->main_base, &tracker->chk_ev);
    event_add(&tracker->chk_ev, NULL);
    return 0;
}

void tracker_handle_write(int sockfd,short event,void *arg)
{
}

void reset_read_buf(struct buffer *buf)
{
    buf->pos = 0;
}
void reset_read_event(int fd,struct connection *entry)
{
    struct thread_worker *worker = entry->worker;
    event_set(&entry->ev,entry->fd,EV_READ,tracker_handle_cmd,entry);
    event_base_set(worker->base,&entry->ev);
    event_add(&entry->ev,NULL);
    return;
}

int checksum(void *buf)
{
}

void tracker_create_file(struct connection *conn,char *p)
{
    struct sockaddr_in peer;
    int len;
    char name[256];
    char* filename;
    struct dfs_metadata *meta = p;
    int sockfd = conn->fd;

    if (getpeername(sockfd,(struct sockaddr*)&peer,&len) < 0) {
        printf("getpeername failed\n");
        return;
    }
    printf("version: %d filesize %d filename %s\n",htonl(meta->version),htonl(meta->filesize),meta->filename);

    snprintf("%d%d%s",peer.sin_addr.s_addr,peer.sin_port,meta->filename);
    filename = MDString(name);
    printf("%s: %s---%s\n",__FUNCTION__,meta->filename,filename);
 
    return;

}

void tracker_register_chunker(struct tracker_struct *tracker,char *p)
{
    struct chunker_register_cmd *cmd = (struct chunk_register_cmd*)p;
    
    struct storage *sg = malloc(sizeof(struct storage));
    assert(sg);
    sg->storageid = ntohl(cmd->storageid);
    strncpy(sg->ip,cmd->ip,sizeof(cmd->ip));
    sg->port = ntohs(cmd->port);
    sg->total = ntohl(cmd->total);
    sg->avail = ntohl(cmd->avail);
    sg->used = ntohl(cmd->used);
    sg->stat = STORAGE_JOIN;
    printf("\nstorageid %d ip %s port %d total %d\n",sg->storageid,sg->ip,sg->port,sg->total);
    insert_storage(&tracker->sg_cluster,sg);
    return;
}


int tracker_get_location(struct connection *conn,char *p)
{
    struct client_getloc_cmd *cmd = (struct client_getloc_cmd*)p;
    
         
    
}

void tracker_response_cmd(struct connection *entry)
{
    struct cadoop_header *head;
    char *p = NULL;
    
    head = (struct cadoop_header *)(entry->rdbuf.buf);
    p = (char*)entry->rdbuf.buf + sizeof(struct cadoop_header);

    switch(head->cmd) {
        case CH_CLIENT_CREATE_FILE:
            tracker_create_file(entry->worker,p);
            break;
        case CH_CLIENT_GET_LOCATION:
            tracker_get_location(entry->worker,p);
        case CH_CLIENT_UPLOAD_FILE:
            break;
        case CH_CLIENT_DOWNLOAD_FILE:
            break;
        case CH_CHUNKER_REGISTER:
            tracker_register_chunker(entry->worker->tracker,p);
            break;
    }
    
}

void reset_buffer(struct buffer *buf)
{
    memset(buf->buf,0,buf->length);
    buf->pos = 0;
}

void tracker_handle_cmd(int sockfd,short event,void *arg)
{  
    struct connection *entry = arg;
    int len = 0;
    struct cadoop_header *head;
    unsigned char type;
    unsigned char cmd;
    unsigned int payload;
    int i;

    printf("tracker_handle_cmd#####\n");
    switch(entry->stat) {
        case CH_TRACKER_WAIT_HEADER:
            printf("recv header\n");
            len = recv(entry->fd,entry->rdbuf.buf + entry->rdbuf.pos,sizeof(struct cadoop_header) - entry->rdbuf.pos,0);
            printf("recv some data\n");
            if (len < 0) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    printf("recv failed %s",strerror(errno));
		    reset_read_event(sockfd,entry);
                }else {
                    printf("recv failed %s",strerror(errno));
                    close(sockfd);
                    return;
                }
            }else if (len == 0) {
                printf("close\n");
                close(sockfd);
                return;
            }
            printf("1:recv len %d\n",len);

            entry->rdbuf.pos += len;
            if (entry->rdbuf.pos == sizeof(struct cadoop_header)) {
                entry->stat = CH_TRACKER_WAIT_DATA;
             }

            reset_read_event(sockfd,entry);
            break;
        case CH_TRACKER_WAIT_DATA:
            printf("recv data\n");
            for (i=0;i<entry->rdbuf.pos;i++) {
                printf("%02x ",entry->rdbuf.buf[i]);
            }
            printf("\n");
#if 1
            head = (struct cadoop_header *)(entry->rdbuf.buf);
            if (MAGIC_NUMBER != ntohl(head->magic)) {
                printf("bad magic\n");
                printf("magic numbger is %x\n",ntohl(head->magic));
                reset_read_buf(&entry->rdbuf);
                return;
            }
            type = head->type;
            cmd = head->cmd;
            payload = ntohl(head->payload);
            printf("magic:%x type %x cmd %x payload %d\n",ntohl(head->magic),type,cmd,payload);
#endif
            len = recv(entry->fd,entry->rdbuf.buf + entry->rdbuf.pos,payload + sizeof(struct cadoop_header) - entry->rdbuf.pos,0);

            if (len < 0) {
                if (errno == EAGAIN) {
                }
                close(sockfd);
            }else if (len == 0) {
                close(sockfd);
            }

            printf("2:recv len %d\n",len);
            entry->rdbuf.pos += len;

            if (payload == entry->rdbuf.pos - sizeof(struct cadoop_header)) {
                //tracker_parse_cmd();
                tracker_response_cmd(entry);
                entry->stat = CH_TRACKER_WAIT_HEADER;
                reset_read_event(sockfd,entry);
                reset_buffer(&entry->rdbuf);
		   	    
            }else {
                reset_read_event(sockfd,entry);
            }
            //entry->stat = CH_TRACKER_HANDLE;
            //reset_read_event(sockfd,entry);
            break;
            case CH_TRACKER_HANDLE:
                tracker_response_cmd(entry);
                entry->stat = CH_TRACKER_WAIT_HEADER;
                reset_read_event(sockfd,entry);
                break;
    }
    
	
}

void tracker_pipe_read(int pipefd, short event, void* arg)
{
    int ret = 0;
    struct thread_worker *worker = arg;
    struct tracker_struct *tracker = worker->tracker;
    struct connection *entry;
    char ch;

    printf("tracker_pipe_read work\n");
    ret = read(pipefd,&ch,1);
    pthread_mutex_lock(&worker->mutex); 
    list_node *first_node = list_remove_first(&worker->connections);
    entry = list_container(first_node,struct connection,node);
    entry->stat = CH_TRACKER_WAIT_HEADER;
    entry->worker = worker;
    pthread_mutex_unlock(&worker->mutex);

    printf("socket %d wait work\n",entry->fd);
    event_set(&entry->ev,entry->fd,EV_READ,tracker_handle_cmd,entry);
    event_base_set(worker->base,&entry->ev);
    event_add(&entry->ev,NULL);
    return;
}

void *tracker_thread_handler(void *arg)
{
    struct thread_worker *info = arg;
    event_base_dispatch(info->base);
}

int create_thread_worker(struct tracker_struct *tracker,int i)
{
    int fd[2];
    int ret = 0;
    struct thread_worker *info = &tracker->pool->threads[i];
    assert(info);
    info->tracker = tracker;
    info->base = event_base_new();
    pthread_mutex_init(&info->mutex,NULL);
    list_init(&info->connections);
    ret = pipe(fd);
    info->notify_read = fd[0];
    info->notify_write = fd[1];
    event_set(&info->notify_ev, fd[0], EV_READ|EV_PERSIST, tracker_pipe_read, info);
    event_base_set(info->base, &info->notify_ev);
    event_add(&info->notify_ev, NULL);

    pthread_create(&info->pid,NULL,tracker_thread_handler,info);
}
struct thread_pool *thread_pool_new(struct tracker_struct *tracker,int size)
{
    int i = 0;
    struct thread_pool *pool = malloc(sizeof(struct thread_pool));
    assert(pool);
    pool->size = (size==0?4:size);
    pool->threads = malloc(pool->size * sizeof(struct thread_worker));
    tracker->pool = pool; 
    for(i=0;i<size;i++) {
      create_thread_worker(tracker,i);
    }
    return pool;

}

int thread_pool_init(struct tracker_struct *tracker,int size)
{
    size = (size==0?4:size);
    thread_pool_new(tracker,size);
    return 0;
}
